package each_webcode

import common.HbaseTools
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.HashMap
import scala.io.Source
import scala.util.parsing.json.JSON;
object transferValueCode_to_Name_eurostat {
        //传入/eurostat_new_conceptid_detailid_detailname.txt   \t
        //返回map(conceptid,map(detailid,detailname))
        def makedic(filepath:String,splitstr:String):mutable.HashMap[String,HashMap[String,String]] ={
                val map04: HashMap[String, HashMap[String, String]] = new HashMap[String, HashMap[String, String]]()
                val datalines = Source.fromFile(filepath).getLines()
                for(dataline <- datalines){
                        val cells = dataline.split(splitstr)
                        if(splitstr=="\t"){

                                if(cells.length==3){
                                        if(map04.contains(cells(0))){
                                                //修改 hashmaps
                                                map04(cells(0)).put(cells(1),cells(2))
                                        }
                                        else{
                                                //新建 hashmap
                                                val tmp: HashMap[String, String] =new  HashMap[String,String]()
                                                tmp.put(cells(1),cells(2))
                                                map04.put(cells(0),tmp)
                                        }
                                }else if(cells.length==2){
                                        if(map04.contains(cells(0))){
                                                //修改 hashmaps
                                                map04(cells(0)).put(cells(1),"")
                                        }
                                        else{
                                                //新建 hashmap
                                                val tmp: HashMap[String, String] =new  HashMap[String,String]()
                                                tmp.put(cells(1),"")
                                                map04.put(cells(0),tmp)
                                        }
                                }else{
                                        print(1)
                                }
                        }
                }
                map04
        }
        //传入/home/hadoop/jars/datas/eurostat_sourcecode_sourcename.txt
        //返回map:(sourcecode,sourcename)
        def makedic(filepath:String):HashMap[String,String]={
                val tmap:HashMap[String,String] = new HashMap[String,String]()
                val datalines = Source.fromFile(filepath).getLines()
                for(dataline<-datalines){
                        val cells = dataline.split("\t")
                        if(cells.length==2){
                                tmap.put(cells(0),cells(1))
                        }
                }
                tmap
        }
        def  isNumber(str:String):Boolean={
                var flag:Boolean = true
                str.foreach(cell=>{
                        if("-0123456789.".indexOf(cell)<0){
                                flag = false
                        }
                })
                flag
        }
        def main(args: Array[String]): Unit = {
                val fatherpath:String = "/home/hadoop/jars/datas"
                //val fatherpath:String ="D:\\InternationalData\\makeChange\\src\\main\\resources\\eurostat"
                val sparkConf = new SparkConf().setAppName("FAOdatachange")//.setMaster("local[*]")
                val sc = new SparkContext(sparkConf)
                // eurostat 比较特殊，没有指定 Value表中数据的对应关系，
                // 读取文件数据，并对数据进行广播
                val new_conceptid_detailid_detailname_path:String = fatherpath + "/eurostat_new_conceptid_detailid_detailname.txt"
                //返回map(conceptid,map(detailid,detailname))
                val new_conceptid = makedic(new_conceptid_detailid_detailname_path,"\t")
                val old_conceptid_detailid_detailname_path:String = fatherpath + "/eurostat_conceptid_detailid_detailname.txt"
                //返回map(conceptid,map(detailid,detailname))
                val old_conceptid = makedic(old_conceptid_detailid_detailname_path, "\t")
                val indicator_dic:String = fatherpath + "/eurostat_sourcecode_sourcename.txt"
                //返回map(sourcecode,sourcename)
                val indicators = makedic(indicator_dic)
                val kbase_dic = fatherpath + "/eurostat_to_kbase.txt"
                //返回map(value字段名,kbase字段名)
                val kbase_indicator = makedic(kbase_dic)
                val value_type_dic = fatherpath + "/eurostat_types.txt"
                //返回map(字段,类型)
                val type_dic = makedic(value_type_dic)
                val conceptid_name = fatherpath + "/eurostat_conceptid_conceptName.txt"
                //map(conceptId,conceptName)
                val conceptid_name_dic = makedic(conceptid_name)

                // 更新了的map(conceptid,map(detailid,detailname)) map04
                val bro_new_conceptId:Broadcast[mutable.HashMap[String, mutable.HashMap[String,String]]]  = sc.broadcast(new_conceptid)
                // 旧版了的map(conceptid,map(detailid,detailname)) map04
                val bro_old_conceptid:Broadcast[mutable.HashMap[String, mutable.HashMap[String,String]]] = sc.broadcast(old_conceptid)
                //旧版新版什么关系?????????????????????????????????????????????
                //map(sourcecode,sourcename) map03
                val bro_indicators:Broadcast[mutable.HashMap[String,String]] = sc.broadcast(indicators)
                //map(value字段名,kbase字段名) map01
                val bro_kbase_indicator:Broadcast[mutable.HashMap[String,String]] = sc.broadcast(kbase_indicator)
                //map(字段,类型)
                val bro_type:Broadcast[mutable.HashMap[String,String]] = sc.broadcast(type_dic)
                //map(conceptId,conceptName) map03
                val bro_dim:Broadcast[mutable.HashMap[String,String]] = sc.broadcast(conceptid_name_dic)
                //本地sub_EUROSTAT_01.json
                val inpath = args(0) //原注释"hdfs://slave1:8020//InternationalData/IMF_Value.json"
                val outpath = args(1) //原注释"hdfs://slave1:8020//InternationalData_change_after/imf_change.value"
                //val numspartion = args(2)
                val baseInfotable: String = new String("BaseMapInfos") //DetailMapInfos
                //增加新的广播变量
                val tools = new HbaseTools()
                val connection = tools.getConnection()
                //参数getFamily_column_value(connection,tableName="BaseMapInfos",rowkey="public",columnFamily="public_Time")
                //返回表名"BaseMapInfos",rowkey="public",columnFamily="public_Time"的所有列名和列值组成的map
                //该表的数据是由/InternationalData/makeDict/src/main/scala/com/cnki/makedic/make_Dic.scala代码写入的
                val PU_timeMap:mutable.HashMap[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_Time")
                //返回表名"BaseMapInfos",rowkey="public",columnFamily="public_country"的所有列名和列值组成的map
                //该表的数据是由/InternationalData/makeDict/src/main/scala/com/cnki/makedic/make_Dic.scala代码写入的
                val PU_country:mutable.HashMap[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_country")
                //special
                connection.close()

                val bro_time:Broadcast[mutable.HashMap[String,String]] = sc.broadcast(PU_timeMap)
                val bro_country:Broadcast[mutable.HashMap[String,String]] = sc.broadcast(PU_country)

                val numspartion = 2
                //该文件存在
                //val inpath = "hdfs://node01:8020//tmp/InternationalData_source/sub_EUROSTAT_Value.json"\
                //该文件存在
                //val outpath = "hdfs://node01:8020//tmp/InternationalData_tmp/sub_eurostat_test_51"

                val webcodes = List("EUROSTAT")
                //make a cache:
                val inputFile = inpath //读取json文件
                val jsonStr = sc.textFile(inputFile) //.sample(true,0.01);

                val result = jsonStr.map(s => JSON.parseFull(s))

                //maps的key是json串的列名,value是json串的列名
                val rdds = result.map {
                        r =>
                                r match {
                                        case Some(maps: Map[String, Any]) => {
                                                val webcode = "EUROSTAT"
                                                //thashmap的key是json串的列名,value是json串的列名
                                                val thashmap: HashMap[String, String] = new HashMap[String, String]()
                                                maps.foreach(t => {
                                                        thashmap.put(t._1.toString , t._2.toString.replace("\n",""))
                                                })
                                                //处理id
                                                thashmap("_id") = thashmap("_id").substring(12, 36)
                                                thashmap("SourceCode") = thashmap("SourceCode").replace(".tsv","").toUpperCase


                                                //处理时间
                                                if(thashmap.contains("\"time")){
                                                        thashmap.put("time",thashmap("\"time").replace("\"",""))
                                                }

                                                var oneLinceMap:HashMap[String,String] = new mutable.HashMap[String,String]()
                                                var dims:Int = 1
                                                var unitDeal:Boolean = false
                                                var areaDeal:Boolean = false
                                                //kv的key是json串的列名,value是json串的列名
                                                thashmap.foreach(kv=>{
                                                        //bro_type:map(字段,类型)
                                                        //如果广播变量bro_type的值中的key包含该字段
                                                        if(bro_type.value.contains(kv._1)){
                                                                //如果广播变量bro_type的值中的value值为"eurostat_type01
                                                                bro_type.value(kv._1) match {
                                                                        //map(kbase字段名,detailname)
                                                                        case "eurostat_type01" =>{
                                                                                // _0: unit ;_1:eurostat_type01
                                                                                //_0 =>
                                                                                //k,v都变成大写
                                                                                val kvs:Tuple2[String,String] = Tuple2(kv._1.toUpperCase,kv._2.trim.toUpperCase)
                                                                                //bro_kbase_indicator:map(value字段名,kbase字段名) map01
                                                                                //bro_new_conceptId:map(conceptid,map(detailid,detailname)) map04
                                                                                //map(value字段名,kbase字段名)中含有key && map(conceptid,map(detailid,detailname))中含有大写的key map(detailid,detailname)含有大写的value
                                                                                if(bro_kbase_indicator.value.contains(kv._1) && bro_new_conceptId.value.contains(kvs._1) && bro_new_conceptId.value(kvs._1).contains(kvs._2))
                                                                                {
                                                                                        //map(kbase字段名,detailname)
                                                                                        oneLinceMap.put(bro_kbase_indicator.value(kv._1),bro_new_conceptId.value(kvs._1)(kvs._2))
                                                                                }
                                                                                else if(bro_kbase_indicator.value.contains(kv._1) && bro_old_conceptid.value.contains(kvs._1) &&bro_old_conceptid.value(kvs._1).contains(kvs._2))
                                                                                {
                                                                                        //map(kbase字段名,detailname)
                                                                                        oneLinceMap.put(bro_kbase_indicator.value(kv._1), bro_old_conceptid.value(kvs._1)(kvs._2))
                                                                                }else{
                                                                                        println(kv)
                                                                                }
                                                                        }
                                                                        //map(kbase字段名,value中的值)
                                                                        case "eurostat_type02" =>{
                                                                                //bro_kbase_indicator:map(value字段名,kbase字段名)
                                                                                if(bro_kbase_indicator.value.contains(kv._1)){  //map01
                                                                                        oneLinceMap.put(bro_kbase_indicator.value(kv._1) , kv._2)
                                                                                }
                                                                        }
                                                                        //用,将key切割成data[2]
                                                                        //map(data[0],大写的value)
                                                                        //map(data[1],detailname)
                                                                        case "eurostat_type03" =>{
                                                                                //大写
                                                                                val kvs:Tuple2[String,String] = Tuple2(kv._1.toUpperCase,kv._2.trim.toUpperCase)
                                                                                //bro_kbase_indicator:map(value字段名,kbase字段名)
                                                                                //将kbase对应的value进行切割
                                                                                val datas = bro_kbase_indicator.value(kv._1).split(",")
                                                                                //通过第一位，获取code，原始值
                                                                                oneLinceMap.put(datas(0), kvs._2)
                                                                                //通过第二位， 转换 concept ,通过detailid 查找 kv._2 获取结果
                                                                                //map(value字段名,kbase字段名)包含该key && map(conceptid,map(detailid,detailname))中含有大写的key map(detailid,detailname)含有大写的value
                                                                                if(bro_kbase_indicator.value.contains(kv._1) && bro_new_conceptId.value.contains(kvs._1) && bro_new_conceptId.value(kvs._1).contains(kvs._2))
                                                                                {
                                                                                        oneLinceMap.put(datas(1) ,bro_new_conceptId.value(kvs._1)(kvs._2))
                                                                                }
                                                                                else if(bro_kbase_indicator.value.contains(kv._1) && bro_old_conceptid.value.contains(kvs._1) &&bro_old_conceptid.value(kvs._1).contains(kvs._2))
                                                                                {
                                                                                        oneLinceMap.put(datas(1),bro_old_conceptid.value(kvs._1)(kvs._2))
                                                                                }else {
                                                                                        println(kv)
                                                                                }
                                                                        }
                                                                        //map(kbase的列,sourcename)
                                                                        //map("SourceName",sourcename)
                                                                        case "eurostat_type04" =>{
                                                                                //需要对数据处理一下，nrg_ti_oilm.tsv
                                                                                //(bro_indicators:map(sourcecode,sourcename)
                                                                                if(bro_indicators.value.contains(kv._2)){//必须有
                                                                                        //map(sourcecode,sourcename)的key中包含kv._2这个sourcecode
                                                                                        oneLinceMap.put(bro_kbase_indicator.value(kv._1) , bro_indicators.value(kv._2))
                                                                                        oneLinceMap.put("SourceName",bro_indicators.value(kv._2))
                                                                                }
                                                                        }
                                                                        //map("DimType",conceptName)
                                                                        //map("Dim1",detailname)
                                                                        case "eurostat_type05" =>{
                                                                                //大写
                                                                                val kvs:Tuple2[String,String] = Tuple2(kv._1.toUpperCase,kv._2.trim.toUpperCase)
                                                                                // bro_dim
                                                                                //dims = 1
                                                                                //dimsN = "Dim1"
                                                                                //dimsT="DimType"
                                                                                val dimsN = "Dim" + dims.toString
                                                                                val dimsT = dimsN + "Type"
                                                                                //bro_dim:map(conceptId,conceptName)
                                                                                if(bro_dim.value.contains(kvs._1)){
                                                                                        oneLinceMap.put(dimsT, bro_dim.value(kvs._1))
                                                                                }else{
                                                                                        println("bad")
                                                                                }
                                                                                //bro_new_conceptId:map(conceptid,map(detailid,detailname))
                                                                                if(bro_new_conceptId.value.contains(kvs._1) && bro_new_conceptId.value(kvs._1).contains(kvs._2))
                                                                                {
                                                                                        oneLinceMap.put(dimsN , bro_new_conceptId.value(kvs._1)(kvs._2))
                                                                                }
                                                                                else if(bro_old_conceptid.value.contains(kvs._1) && bro_old_conceptid.value(kvs._1).contains(kvs._2))
                                                                                {
                                                                                        oneLinceMap.put(dimsN,bro_old_conceptid.value(kvs._1)(kvs._2))
                                                                                }else{
                                                                                        //println("eurostat_type05")
                                                                                        //println(kv)
                                                                                        oneLinceMap.put(dimsN,"")
                                                                                }
                                                                                dims += 1
                                                                        }
                                                                        //按照" "切割
                                                                        //长度为1时
                                                                        //map(kbase的列名,value的列值)
                                                                        //长度为2时,切割为data[2]
                                                                        //map(kbase的列名,data[0])
                                                                        //map("FootNote",detailname)
                                                                        case "eurostat_type06" =>{
                                                                                //bro_kbase_indicator:map(value字段名,kbase字段名)
                                                                                if(bro_kbase_indicator.value.contains(kv._1)){
                                                                                        //按" "空格拆分数据,
                                                                                        val datas = kv._2.split(" ")
                                                                                        if(datas.length==1){
                                                                                                oneLinceMap.put(bro_kbase_indicator.value(kv._1), kv._2.replace(":",""))

                                                                                        }else if(datas.length==2){
                                                                                                // datas[1] ,使用 obs_status 进行数据的转换处理。
                                                                                                if(isNumber(datas(0))){
                                                                                                        oneLinceMap.put(bro_kbase_indicator.value(kv._1),datas(0))
                                                                                                        //通过第二部分去查找 conceptid,conceptname,用于说明
                                                                                                        //map(conceptid,map(detailid,detailname))
                                                                                                        if(bro_old_conceptid.value("OBS_STATUS").contains(datas(1))){
                                                                                                                oneLinceMap.put("FootNote",bro_old_conceptid.value("OBS_STATUS")(datas(1)))
                                                                                                        }else{
                                                                                                                val status = ""
                                                                                                                val all_status = datas(1).map(cell=>{
                                                                                                                        if(bro_old_conceptid.value("OBS_STATUS").contains(cell.toString)){
                                                                                                                                bro_old_conceptid.value("OBS_STATUS")(cell.toString)
                                                                                                                        }else{
                                                                                                                                ""
                                                                                                                        }
                                                                                                                })

                                                                                                                oneLinceMap.put("FootNote",all_status.mkString(", "))
                                                                                                        }
                                                                                                }
                                                                                        }else{
                                                                                                println(1)
                                                                                        }
                                                                                }
                                                                        }
                                                                        case "eurostat_type07" =>{
                                                                                //unitDeal=false
                                                                                if(!unitDeal){
                                                                                        val kvs:Tuple2[String,String] = Tuple2(kv._1.toUpperCase,kv._2.trim.toUpperCase)
                                                                                        //橙黄色 currency/unit
                                                                                        val chengHuangse = Array("LC_N00COST","LC_RCOST_R2","LC_NCOST_R2","LC_NCOSTOT_R2","NRG_PC_202_C","NRG_PC_203_C","NRG_PC_202","NRG_PC_202_H","NRG_PC_203","NRG_PC_203_H","NRG_PC_204","NRG_PC_204_H","NRG_PC_205","NRG_PC_205_H","NRG_PC_206_H","TEN00118","TEN00117","LC_R00COST")
                                                                                        //天蓝色  currency对应kbase中的Unit；unit对应kbase中的ReferencePeriod
                                                                                        val tianlanse =Array("FOR_EAF03")
                                                                                        //橙红色  kbase中Unit为固定值%。Currency可省略
                                                                                        val chengHongse = Array("ext_lt_invcur","tet00067","tet00068")
                                                                                        //紫色 indic_il取share的时候，kbase中Unit为固定值%，Currency可省略；indic_il取TC的时候，kbase中Unit为currency。
                                                                                        val zise = Array("ILC_DI01")
                                                                                        //蓝色  currency是一个指标维度，unit是单位。
                                                                                        val langse = Array("GOV_10DD_DCUR")

                                                                                        //先获取数据，因为需要从 code 到 name的转换，所以这里就先处理。后续根据情况进行组装
                                                                                        var  unitname = ""
                                                                                        var currency_name = ""
                                                                                        //bro_old_conceptid:map(conceptid,map(detailid,detailname))
                                                                                        if(thashmap.contains("unit") && bro_old_conceptid.value.contains("UNIT") &&bro_old_conceptid.value("UNIT").contains(thashmap("unit"))) {
                                                                                                unitname = bro_old_conceptid.value("UNIT")(thashmap("unit"))
                                                                                        }
                                                                                        if(thashmap.contains("currency") && bro_old_conceptid.value.contains("CURRENCY") &&bro_old_conceptid.value("CURRENCY").contains(thashmap("currency"))) {
                                                                                                currency_name = bro_old_conceptid.value("CURRENCY")(thashmap("currency"))
                                                                                        }

                                                                                        if(chengHuangse.contains(thashmap("SourceCode"))){
                                                                                                //组合 currency 和 unit
                                                                                                oneLinceMap.put("Unit",currency_name+"/"+unitname)
                                                                                        }else if(tianlanse.contains(thashmap("SourceCode"))){
                                                                                                oneLinceMap.put("Unit",currency_name)
                                                                                                oneLinceMap.put("ReferencePeriod",unitname)
                                                                                        }else if(chengHongse.contains(thashmap("SourceCode"))){
                                                                                                oneLinceMap.put("Unit",unitname)
                                                                                        }else if(zise.contains(thashmap("SourceCode"))){
                                                                                                if(thashmap.contains("indic_il") && thashmap("indic_il")=="share"){
                                                                                                        oneLinceMap.put("Unit","%")
                                                                                                }else if(thashmap.contains("indic_il") && thashmap("indic_il")=="TC"){
                                                                                                        oneLinceMap.put("Unit",currency_name)
                                                                                                }
                                                                                        }else if(langse.contains(thashmap("SourceCode"))){
                                                                                                val dimsN = "Dim" + dims.toString
                                                                                                val dimsT = dimsN + "Type"
                                                                                                oneLinceMap.put("Unit",unitname)
                                                                                                oneLinceMap.put(dimsT,"Currency")
                                                                                                oneLinceMap.put(dimsN,currency_name)
                                                                                                dims += 1
                                                                                        }else if(thashmap.contains("currency")){//currency,对应kbase中的Unit)
                                                                                                oneLinceMap.put("Unit",currency_name)
                                                                                        }else{//只含有unit
                                                                                                oneLinceMap.put("Unit",unitname)
                                                                                        }
                                                                                        unitDeal = true
                                                                                }
                                                                        }
                                                                        //map()很多字段,直接取值处理
                                                                        //map("Period",四位年)
                                                                        //map("year",四位年)
                                                                        case "eurostat_type08"=>{
                                                                                if(!areaDeal){
                                                                                        //seabasin与rep_mar同时出现时，seabasin，rep_mar 拼接
                                                                                        val  sr =Array("MAR_SG_AM_CWS")
                                                                                        //seabasin 单独出现
                                                                                        val s =Array("MAR_SG_AM_EWX")
                                                                                        //seabasin 和 c_regis 同时出现，seabasin--c_regis对应kbase的Reporter
                                                                                        val sc = Array("TRAN_SF_MARV","TRAN_SF_MARVES","TRAN_SF_MARVPER")
                                                                                        //c_regis与geo同时出现,c_regis对应kbase的Parterner，geo对应reporter
                                                                                        val cg = Array("TRAN_SF_AVIAAW","TRAN_SF_AVIACA","TRAN_SF_AVIAGAH","TRAN_SF_AVIAGAL")

                                                                                        var rep_mar = ""
                                                                                        var rep_mar_name = ""
                                                                                        if(thashmap.contains("rep_mar")){
                                                                                                rep_mar = thashmap("rep_mar")
                                                                                                if(bro_new_conceptId.value("REP_MAR").contains(rep_mar)){
                                                                                                        rep_mar_name = bro_new_conceptId.value("REP_MAR").getOrElse(rep_mar,"")
                                                                                                }else{
                                                                                                        rep_mar_name = bro_old_conceptid.value("REP_MAR").getOrElse(rep_mar,"")
                                                                                                }

                                                                                        }

                                                                                        var seabasin = ""
                                                                                        var seabasin_name = ""
                                                                                        if(thashmap.contains("seabasin")){
                                                                                                seabasin = thashmap("seabasin")
                                                                                                seabasin_name = bro_old_conceptid.value("SEABASIN").getOrElse(seabasin,"")
                                                                                        }

                                                                                        var c_regis = ""
                                                                                        var c_regis_name = ""
                                                                                        if(thashmap.contains("c_regis")){
                                                                                                c_regis = thashmap("c_regis")
                                                                                                c_regis_name = bro_old_conceptid.value("C_REGIS").getOrElse(c_regis,"")
                                                                                        }

                                                                                        var geo = ""
                                                                                        var geo_name = ""
                                                                                        if(thashmap.contains("geo")){
                                                                                                geo = thashmap("geo")
                                                                                                if(bro_new_conceptId.value("REP_MAR").contains(rep_mar)){
                                                                                                        geo_name = bro_new_conceptId.value("GEO").getOrElse(geo,"")
                                                                                                }else{
                                                                                                        geo_name = bro_old_conceptid.value("GEO").getOrElse(geo,"")
                                                                                                }
                                                                                        }

                                                                                        if(sr.contains(thashmap("SourceCode"))){//seabasin，rep_mar 拼接
                                                                                                oneLinceMap.put("ReporterCode",seabasin+","+rep_mar)
                                                                                                oneLinceMap.put("Reporter",seabasin_name+","+rep_mar_name)
                                                                                        }else if(s.contains(thashmap("SourceCode"))){//seabasin 单独出现
                                                                                                oneLinceMap.put("ReporterCode",seabasin)
                                                                                                oneLinceMap.put("Reporter",seabasin_name)
                                                                                        }else if(sc.contains(thashmap("SourceCode"))){ //seabasin 和 c_regis 同时出现，seabasin--c_regis对应kbase的Reporter
                                                                                                oneLinceMap.put("ReporterCode",seabasin+","+c_regis)
                                                                                                oneLinceMap.put("Reporter",seabasin_name+","+c_regis_name)
                                                                                        }else if(cg.contains(thashmap("SourceCode"))){ //c_regis与geo同时出现,c_regis对应kbase的Parterner，geo对应reporter
                                                                                                oneLinceMap.put("ReporterCode",geo)
                                                                                                oneLinceMap.put("Reporter",geo_name)
                                                                                                oneLinceMap.put("ParternerCode",c_regis)
                                                                                                oneLinceMap.put("Parterner",c_regis_name)
                                                                                        }else{
                                                                                                oneLinceMap.put("ReporterCode",geo)
                                                                                                oneLinceMap.put("Reporter",geo_name)
                                                                                        }
                                                                                        areaDeal = true
                                                                                }
                                                                        }
                                                                }
                                                        }
                                                })
                                                oneLinceMap.put("SourceCode",thashmap("SourceCode"))
                                                oneLinceMap.put("WebCode","EUROSTAT")
                                                oneLinceMap.put("WebName","欧盟统计局")

                                                //合并字段，生成
                                                //检测 Year字段，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap("Year")!=""){
                                                        if(oneLinceMap("Year").length>4 && !oneLinceMap.contains("Period")){
                                                                oneLinceMap.put("Period",oneLinceMap("Year").substring(4))
                                                                oneLinceMap("Year") = oneLinceMap("Year").substring(0,4)
                                                        }
                                                        //检测Period字段
                                                        //1检查是否有值，
                                                }
                                                //生成TimeSort字段；Period字段判断是否存在并进行转换，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap.contains("Period")&&oneLinceMap("Period").trim!=""){
                                                        if(PU_timeMap.contains(oneLinceMap("Period").trim)){
                                                                //PU_timeMap:表名"BaseMapInfos",rowkey="public",columnFamily="public_Time"的所有列名和列值组成的map
                                                                oneLinceMap.put("TimeSort",oneLinceMap("Year")+PU_timeMap(oneLinceMap("Period").trim))
                                                        }else{
                                                                oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                        }
                                                }else{
                                                        oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                }
                                                //调试
                                                if(!oneLinceMap.contains("TimeSort")){
                                                        print(1)
                                                }else{
                                                        if(oneLinceMap("TimeSort").replace(" ","")==""){
                                                                if(!oneLinceMap.contains("Year")){
                                                                        print(oneLinceMap("ObjectId"))
                                                                }
                                                        }
                                                }
                                                //生成FuzzyIndicator字段
                                                if(oneLinceMap.contains("Indicator")&&oneLinceMap("Indicator")!="")
                                                {oneLinceMap.put("FuzzyIndicator",oneLinceMap("Indicator"))
                                                        //生成IndicatorLength字段
                                                        oneLinceMap.put("IndicatorLength",oneLinceMap("Indicator").length.toString)
                                                }
                                                //生成FuzzyReporter字段
                                                if(oneLinceMap.contains("Reporter")&&oneLinceMap("Reporter")!=""){
                                                        if(bro_country.value.contains(oneLinceMap("Reporter"))){
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter")+"，"+bro_country.value(oneLinceMap("Reporter")))
                                                        }else{
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter"))
                                                        }
                                                }
                                                //Parterner
                                                //生成FUZZYPARTERNER字段
                                                if(oneLinceMap.contains("Parterner")&&oneLinceMap("Parterner")!=""){
                                                        if(bro_country.value.contains(oneLinceMap("Parterner"))){
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner")+"，"+bro_country.value(oneLinceMap("Parterner")))
                                                        }else{
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner"))
                                                        }
                                                }
                                                val MetaData:Array[String] = Array("WebCode", "SourceCode", "ReporterCode", "Reporter", "ParternerCode", "Parterner", "Year", "Period",
                                                        "IndicatorCategory", "Indicator", "Dim1Type", "Dim1", "Dim2Type", "Dim2", "Dim3Type", "Dim3",
                                                        "Dim4Type", "Dim4", "Dim5Type", "Dim5", "Dim6Type", "Dim6", "Dim7Type", "Dim7", "Dim8Type", "Dim8" )
                                                val DIMTYPEMETADATA:Array[String] =Array("Dim1Type", "Dim2Type", "Dim3Type", "Dim4Type", "Dim5Type", "Dim6Type", "Dim7Type", "Dim8Type")
                                                val DIMMETADATA:Array[String] = Array("Dim1", "Dim2", "Dim3", "Dim4", "Dim5", "Dim6", "Dim7", "Dim8")

                                                //生成DIMTYPEMETADATA
                                                var dimtypemetadata = DIMTYPEMETADATA.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")

                                                //生成DIMMETADATA
                                                var dimemetadata = DIMMETADATA.flatMap(
                                                        onecel=>{
                                                                oneLinceMap.getOrElse(onecel,"<->")+"，"
                                                        }
                                                ).mkString("").replace("<->，","")
                                                //生成MetaData
                                                var metadata = MetaData.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")
                                                oneLinceMap.put("DIMTYPEMETADATA",dimtypemetadata)
                                                oneLinceMap.put("DIMMETADATA",dimemetadata)
                                                oneLinceMap.put("MetaData",metadata)
                                                //val kkl:RDD[String] = sc.parallelize(().toString)
                                                oneLinceMap.mkString("(<>)").replace(" -> ","\t")
                                        }
                                }
                }.saveAsTextFile(outpath)

        }
}
